package drds.binlog.parse.index;

import com.google.common.base.Function;
import com.google.common.collect.ComputingMap;
import drds.binlog.common.position.LogPosition;
import drds.binlog.common.utils.JsonUtils;
import drds.binlog.metadata.MetaDataManagerException;
import drds.binlog.parse.exception.BinlogParseException;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 基于文件刷新的log
 * position实现
 * <p>
 * <pre>
 * 策略：
 * 1. 先写内存，然后定时刷新数据到File
 * 2. 数据采取overwrite模式(只保留最后一次)
 * </pre>
 */
public class FileMixedLogPositionManager extends AbstractLogPositionManager
{

    private final static Logger logger = LoggerFactory.getLogger(FileMixedLogPositionManager.class);
    private final static Charset charset = Charset.forName("UTF-8");
    @SuppressWarnings("serial")
    private final LogPosition nullPosition = new LogPosition()
    {
    };
    private File dataDir;
    private Map<String, File> dataFileCaches;
    private ScheduledExecutorService scheduledExecutorService;
    private MemoryLogPositionManager memoryLogPositionManager;

    private long period;
    private Set<String> persistTasks;

    public FileMixedLogPositionManager(File dataDir, long period, MemoryLogPositionManager memoryLogPositionManager)
    {
        if (dataDir == null)
        {
            throw new NullPointerException("null dataDir");
        }
        if (period <= 0)
        {
            throw new IllegalArgumentException("period must be positive, given: " + period);
        }
        if (memoryLogPositionManager == null)
        {
            throw new NullPointerException("null memoryLogPositionManager");
        }
        this.dataDir = dataDir;
        this.period = period;
        this.memoryLogPositionManager = memoryLogPositionManager;

        this.dataFileCaches = ComputingMap.makeComputingMap(new Function<String, File>()
        {

            public File apply(String destination)
            {
                return getDataFile(destination);
            }
        });

        this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
        this.persistTasks = Collections.synchronizedSet(new HashSet<String>());
    }

    @Override
    public void start()
    {
        super.start();

        if (!dataDir.exists())
        {
            try
            {
                FileUtils.forceMkdir(dataDir);
            } catch (IOException e)
            {
                throw new MetaDataManagerException(e);
            }
        }

        if (!dataDir.canRead() || !dataDir.canWrite())
        {
            throw new MetaDataManagerException("dir[" + dataDir.getPath() + "] can not read/write");
        }

        if (!memoryLogPositionManager.isRunning())
        {
            memoryLogPositionManager.start();
        }

        // 启动定时工作任务
        scheduledExecutorService.scheduleAtFixedRate(new Runnable()
        {

            public void run()
            {
                List<String> tasks = new ArrayList<String>(persistTasks);
                for (String destination : tasks)
                {
                    try
                    {
                        // 定时将内存中的最新值刷到file中，多次变更只刷一次
                        flushDataToFile(destination);
                        persistTasks.remove(destination);
                    } catch (Throwable e)
                    {
                        // ignore
                        logger.error("period update" + destination + " curosr failed!", e);
                    }
                }
            }
        }, period, period, TimeUnit.MILLISECONDS);

    }

    @Override
    public void stop()
    {
        super.stop();

        flushDataToFile();
        scheduledExecutorService.shutdown();
        memoryLogPositionManager.stop();
    }

    @Override
    public LogPosition getLatestIndexBy(String destination)
    {
        LogPosition logPosition = memoryLogPositionManager.getLatestIndexBy(destination);
        if (logPosition != null)
        {
            return logPosition;
        }
        logPosition = loadDataFromFile(dataFileCaches.get(destination));
        if (logPosition == null)
        {
            return nullPosition;
        }
        return logPosition;
    }

    @Override
    public void persistLogPosition(String destination, LogPosition logPosition) throws BinlogParseException
    {
        persistTasks.add(destination);
        memoryLogPositionManager.persistLogPosition(destination, logPosition);
    }

    // ============================ helper method ======================

    private File getDataFile(String destination)
    {
        File destinationMetaDir = new File(dataDir, destination);
        if (!destinationMetaDir.exists())
        {
            try
            {
                FileUtils.forceMkdir(destinationMetaDir);
            } catch (IOException e)
            {
                throw new MetaDataManagerException(e);
            }
        }

        String dataFileName = "parse.dat";
        return new File(destinationMetaDir, dataFileName);
    }

    private void flushDataToFile()
    {
        for (String destination : memoryLogPositionManager.destinations())
        {
            flushDataToFile(destination);
        }
    }

    private void flushDataToFile(String destination)
    {
        flushDataToFile(destination, dataFileCaches.get(destination));
    }

    private void flushDataToFile(String destination, File dataFile)
    {
        LogPosition position = memoryLogPositionManager.getLatestIndexBy(destination);
        if (position != null && position != nullPosition)
        {
            String json = JsonUtils.marshalToString(position);
            try
            {
                FileUtils.writeStringToFile(dataFile, json);
            } catch (IOException e)
            {
                throw new MetaDataManagerException(e);
            }
        }
    }

    private LogPosition loadDataFromFile(File dataFile)
    {
        try
        {
            if (!dataFile.exists())
            {
                return null;
            }

            String json = FileUtils.readFileToString(dataFile, charset.name());
            return JsonUtils.unmarshalFromString(json, LogPosition.class);
        } catch (IOException e)
        {
            throw new MetaDataManagerException(e);
        }
    }
}
